package com.oracle.violet.first.report
import com.oracle.violet.first.config.ConfigHandler
import com.oracle.violet.first.until.C3P0Pools
import org.apache.commons.dbutils.QueryRunner
import org.apache.spark.sql.{DataFrame, SQLContext, SaveMode}
import org.apache.spark.{SparkConf, SparkContext}

object TermUnit {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf()
      .setAppName("地域分布")
      .setMaster("local[*]")
      .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
    val sc = new SparkContext(conf)
    val sQLContext = new SQLContext(sc)

    //读取文件
    val dataFrame: DataFrame = sQLContext.read.parquet(ConfigHandler.parquetFilePath)
    import sQLContext.implicits._
    //维度
    val sourceCode = dataFrame.map(row => {
      val ispname = row.getAs[String]("ispname") //运营商
      val networkmannername = row.getAs[String]("networkmannername") //网络类型
      val devicetype = row.getAs[Int]("devicetype") //设备类型
      val client = row.getAs[Integer]("client") //操作系统

      //提取相应字段
      val reqmode = row.getAs[Int]("requestmode")
      val pronode = row.getAs[Int]("processnode")
      val iseffect = row.getAs[Int]("iseffective")
      val isbiling = row.getAs[Int]("isbilling")
      val isbid = row.getAs[Int]("isbid")
      val iswin = row.getAs[Int]("iswin")
      val adorderid = row.getAs[Int]("adorderid")


      val adRawReq = if (reqmode == 1 && pronode >= 1) 1 else 0
      val adEffReq = if (reqmode == 1 && pronode >= 2) 1 else 0
      val adReq = if (reqmode == 1 && pronode == 3) 1 else 0

      val adRtbsReq = if (iseffect == 1 && isbiling == 1 && isbid == 1 && adorderid != 0) 1 else 0
      val adRtbReq = if (iseffect == 1 && isbiling == 1 && iswin == 1) {
        val winprice = row.getAs[Double]("winprice")
        val adpayment = row.getAs[Double]("adpayment")
        (1, winprice / 1000, adpayment / 1000)
      } else (0, 0d, 0d)
      val adshow = if (reqmode == 2 && iseffect == 1) 1 else 0
      val adClick = if (reqmode == 3 && iseffect == 1) 1 else 0

      //组装成元组
      ((ispname, networkmannername, devicetype, client), List(adRawReq, adEffReq, adReq, adRtbsReq, adRtbReq._1, adshow, adClick, adRtbReq._2, adRtbReq._3))
    }).cache()


    //运营商
     sourceCode.map(x=>(x._1._1,x._2)).reduceByKey((a, b)=>a.zip(b).map(x=>x._1+x._2)).toDF().write.mode(SaveMode.Overwrite).jdbc(ConfigHandler.url, "orc_termunit_operator", ConfigHandler.props)

     //网络类型
    sourceCode.map(x=>(x._1._2,x._2)).reduceByKey((a, b)=>a.zip(b).map(x=>x._1+x._2)).toDF().write.mode(SaveMode.Overwrite).jdbc(ConfigHandler.url, "orc_termunit_nettype", ConfigHandler.props)
    //设备类型
    sourceCode.map(x=>(x._1._3,x._2)).reduceByKey((a, b)=>a.zip(b).map(x=>x._1+x._2)).toDF().write.mode(SaveMode.Overwrite).jdbc(ConfigHandler.url, "orc_termunit_unittype", ConfigHandler.props)
    //操作系统
    sourceCode.map(x=>(x._1._4,x._2)).reduceByKey((a, b)=>a.zip(b).map(x=>x._1+x._2)) .foreachPartition(partition=>{
      val runner = new QueryRunner(C3P0Pools.getDataSource)
      val params = partition.map(x=>Array(x._1,x._2(0).asInstanceOf[AnyRef],x._2(1).asInstanceOf[AnyRef])).toArray
      runner.batch("replace into t_mydata values(?,?,?)",params)
    })
  sc.stop()






  }

}
